package SparkStreaming;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class PersistWordCount {

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("PersistWordCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

        jssc.checkpoint("D;/wordcount");

        //wordCount基础逻辑
        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);

        JavaDStream<String> words = lines.flatMap((FlatMapFunction<String, String>) line -> (Iterator<String>) Arrays.asList(line.split(" ")));
        JavaPairDStream<String, Integer> pairs = words.mapToPair((PairFunction<String, String, Integer>) word -> new Tuple2<String, Integer>(word, 1));
        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((Function2<Integer, Integer, Integer>) (integer, integer2) -> integer + integer2);
        //每次得到当前所有单词的统计次数之后，将其写入mysql存储，进行持久化，以便于后续的其他操作

        //wordCountsRDD
       /* wordCounts.foreachRDD((VoidFunction2<JavaPairRDD<String, Integer>, Void>) wordsCountsRDD -> {
            wordsCountsRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Integer>>>() {
                @Override
                public void call(Iterator<Tuple2<String, Integer>> wordCounts1) throws Exception {
                    //给每个partition，获取一个连接
                    Connection conn = ConnectionPool.getConnection();
                    //遍历partition中的数据，使用一个连接，插入数据库
                    Tuple2<String, Integer> wordCount = null;
                    while (wordCounts1.hasNext()) {
                        wordCount = wordCounts1.next();
                        String sql = "insert into wordcount(word,count) values(" + wordCount._1 + "," + wordCount._2 + ")";
                        Statement stmt = conn.createStatement();
                        stmt.executeUpdate(sql);
                    }

                    //用完以后，将连接还回去
                    ConnectionPool.returnConnection(conn);
                }
            });
            return null;
        });*/

        /*jssc.start();
        jssc.awaitTermination();
        jssc.close();*/
    }
}
